package com.icreate.storm.methods.TupleAccumulate.indicator;

import com.icreate.storm.methods.TupleAccumulate.TupleAccumulator;
import com.icreate.entity.indicator.IndicatorID;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.ITuple;
import org.bson.Document;

import java.util.LinkedHashMap;
import java.util.Map;
import java.util.Set;

/**
 * Created by feng on 2017/5/14.
 */
public class IndicatorAccumulator implements TupleAccumulator {
    private Set<String> excludeFields;
    private Map<String, Long> counter = new LinkedHashMap<>();

    public IndicatorAccumulator(Set<String> excludeFields) {
        this.excludeFields = excludeFields;
    }

    @Override
    public void add(ITuple tuple) {
        IndicatorID indicatorID = (IndicatorID) tuple.getValueByField("IndicatorID");
        String hour = indicatorID.getHour();
        Fields fields = tuple.getFields();
        for (String field : fields) {
            if (!excludeFields.contains(field))
                add("hour." + hour + "." + field, (Long) tuple.getValueByField(field));
        }
    }

    @Override
    public Document toDocument() {
        Document document = new Document();
        Document incDocument = new Document();
        for (Map.Entry<String, Long> entry : this.counter.entrySet()) {
            incDocument.append(entry.getKey(), entry.getValue());
        }
        document.append("$inc", incDocument);
        return document;
    }

    @Override
    public void reset() {

    }

    private void add(String key, Long value) {
        counter.put(key, counter.getOrDefault(key, 0L) + value);
    }
}
